package com.lightbend.akka.failures

import java.util.concurrent.atomic.AtomicInteger

import akka.actor.{ActorSystem, Scheduler}
import akka.stream.scaladsl._
import akka.stream._

import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutor, Future}
import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.{Failure, Success, Try}

object NonLinearGraph extends App {

  private implicit val system          : ActorSystem              = ActorSystem("NonLinearGraph")
  private implicit val materializer    : ActorMaterializer        = ActorMaterializer()
  private implicit val executionContext: ExecutionContextExecutor = system.dispatcher

  Source(List("1", "2", "a", "4", "5"))
  .map(_.toInt)
  .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
  .runForeach(println)

  val nonLinearFlow = Flow.fromGraph(GraphDSL.create() { implicit builder =>
    import GraphDSL.Implicits._
    val dangerousFlow = Flow[Int].map {
      case 5 => throw new RuntimeException("BOOM!")
      case x => x
    }

    val safeFlow = Flow[Int]
    val bcast = builder.add(Broadcast[Int](2))
    val zip = builder.add(Zip[Int, Int])

    bcast ~> dangerousFlow ~> zip.in0
    bcast ~> safeFlow ~> zip.in1

    FlowShape(bcast.in, zip.out)
  })

  Source(1 to 10)
  .via(nonLinearFlow)
  .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
  .runForeach(println)

  def makeFuture(r: Int): Future[Int] = {
    Future {
      if (r == 5) {
        Thread.sleep(20)
        throw new RuntimeException("BOOM!")
      }
      else
        Thread.sleep(500)
      r
    }
  }
  val futureFlow = Flow.fromGraph(GraphDSL.create() { implicit builder =>
    import GraphDSL.Implicits._

    val dangerousFlow = Flow[Int].mapAsyncUnordered[Int](5)(makeFuture)

    val safeFlow = Flow[Int]
    val printFlow = builder.add(Flow[Int])
    val bcast = builder.add(Broadcast[Int](2)) // broadcast 由于使用back-pressure机制，执行时是按照顺序执行的
  val zip = builder.add(Zip[Int, Int])

    bcast ~> dangerousFlow ~> printFlow ~> zip.in0
    bcast ~> safeFlow ~> zip.in1

    FlowShape(bcast.in, zip.out)
  })

  Source(1 to 10)
  .via(futureFlow)
  .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
  .runForeach(println)

  val futureFlow1 = Flow.fromGraph(GraphDSL.create() { implicit builder =>
    import GraphDSL.Implicits._

    val dangerousFlow = Flow[Int].map {
      case 5 => Failure(new RuntimeException("BOOM!"))
      case x => Try(x)
    }

    val safeFlow = Flow[Int].map(Try(_))
    val bcast = builder.add(Broadcast[Int](2))
    val zip = builder.add(Zip[Try[Int], Try[Int]])

    bcast ~> dangerousFlow ~> zip.in0
    bcast ~> safeFlow ~> zip.in1

    FlowShape(bcast.in, zip.out)
  })

  Source(1 to 10)
  .via(futureFlow1)
  .collect {
    case (Success(a), Success(b)) => (a, b)
  }
  .runWith(Sink.foreachParallel(3)(println))

  Source(1 to 10)
  .mapAsyncUnordered(4)(makeFuture)
  .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
  .runForeach(elem => println(s"after: $elem"))


}

